package com.niiwoo.civet.trade.sharding;

import java.util.Collection;
import java.util.LinkedHashSet;

import com.dangdang.ddframe.rdb.sharding.api.ShardingValue;
import com.dangdang.ddframe.rdb.sharding.api.strategy.table.SingleKeyTableShardingAlgorithm;
import com.google.common.collect.Range;

public class TradeByUserIdTableShardingAlgorithm implements SingleKeyTableShardingAlgorithm<String> {

    @Override
    public String doEqualSharding(final Collection<String> availableTargetNames, final ShardingValue<String> shardingValue) {
        String shardIndex = "_" + Math.abs(Long.valueOf(shardingValue.getValue().hashCode())) % availableTargetNames.size();
        for (String each : availableTargetNames) {
            if (each.endsWith(shardIndex)) {
                return each;
            }
        }
        throw new UnsupportedOperationException();
    }

    @Override
    public Collection<String> doInSharding(final Collection<String> availableTargetNames, final ShardingValue<String> shardingValue) {
        Collection<String> result = new LinkedHashSet<>(availableTargetNames.size());
        Collection<String> values = shardingValue.getValues();
        for (String value : values) {
            long shardIndex = Math.abs(Long.valueOf(value.hashCode())) % availableTargetNames.size();
            for (String each : availableTargetNames) {
                if (each.endsWith("_" + shardIndex)) {
                    result.add(each);
                }
            }
        }
        return result;
    }
    
    @Override
    public Collection<String> doBetweenSharding(final Collection<String> availableTargetNames, final ShardingValue<String> shardingValue) {
        Collection<String> result = new LinkedHashSet<>(availableTargetNames.size());
        Range<String> range = shardingValue.getValueRange();

        long startValue = Math.abs(Long.valueOf(range.lowerEndpoint().hashCode()));
        long endValue = Math.abs(Long.valueOf(range.upperEndpoint().hashCode()));

        for (long value = startValue; value <= endValue; value++) {
            long shardIndex = value % availableTargetNames.size();
            for (String each : availableTargetNames) {
                if (each.endsWith("_" + shardIndex)) {
                    result.add(each);
                }
            }
        }
        return result;
    }
}